package com.aha.dataanalyzeandview.utils;


import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.thrift.TException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;


/**
 * hive 元数据操作
 */
@Service
@Slf4j
public class HiveDataOperatorUtils {


    /**
     * 初始化 HiveMetaStore 连接
     *
     * @return
     */
    public HiveMetaStoreClient initHiveMetaStoreClient(String hiveMetastoreUri) {

        HiveMetaStoreClient hiveMetaStoreClient = null;
        HiveConf hiveConf = new HiveConf();
        System.setProperty("HADOOP_USER_NAME", "hdfs");

        hiveConf.set("hive.metastore.uris", hiveMetastoreUri);
        try {
            hiveMetaStoreClient = new HiveMetaStoreClient(hiveConf);
            return hiveMetaStoreClient;
        } catch (MetaException e) {
            e.printStackTrace();
        }

        return null;
    }


    /**
     * 删除指定的表分区数据
     *
     * @param hiveMetaStoreClient
     */
    public boolean delHivePartitionData(HiveMetaStoreClient hiveMetaStoreClient, String dbName, String tableName, List<String> retainPartitionsStrList) {
        boolean b = false;
        try {
            b = hiveMetaStoreClient.dropPartition(
                    dbName,tableName, retainPartitionsStrList);
            if (!b) {
                log.error("删除分区数据失败！执行删除的hive库名是：【{}】，表名是：【{}】",
                        dbName, tableName);

            } else {
                log.info("删除分区数据成功！执行删除的hive库名是：【{}】，表名是：【{}】",
                        dbName, tableName);
            }
        } catch (TException e) {
            e.printStackTrace();
        }

        return b;

    }


    /**
     * 获取hive表所有的分区字段
     *
     * @return
     */
    public List<FieldSchema> getHiveTableColumns(HiveMetaStoreClient hiveMetaStoreClient, String dbName, String tableName) {
        try {
            Table table = hiveMetaStoreClient.getTable(dbName, tableName);
            List<FieldSchema> cols = table.getSd().getCols();

            return cols;
        } catch (TException e) {
            e.printStackTrace();
        }

        return null;
    }


    /**
     * 关闭hivemetastore 的连接
     *
     * @param hiveMetaStoreClient
     */
    public void closeHiveMetaStore(HiveMetaStoreClient hiveMetaStoreClient) {
        if (hiveMetaStoreClient != null) {
            hiveMetaStoreClient.close();
        }
    }


    /**
     * @param retainPartition 需要保留的分区的值
     */
    public void deleteHivePartitionData(String hiveMetastoreUri,String hiveJdbcUrl,String dbName, String tableName, Set<String> retainPartition) {

//        String hiveJdbcUrl = "jdbc:hive2://192.168.10.132:10000";
        String sql = String.format("show partitions %s.%s", dbName, tableName);

//        Map<String,Object> map = new HashMap<>();
//        map.put("partdate",String.class);
//        List<Map<String, Object>> list = new ArrayList<>();
//        list.add(map);

        List<String> partitionStrList = new ArrayList<>();
        List<List<String>> lists = MyJdbcUtils.executeQuery(hiveJdbcUrl, "",
                dbName, "", "",
                null, sql, 1);

//        for (Map<String, Object> m : res) {
//            for (Map.Entry<String, Object> kv : m.entrySet()) {
//                String partitionValue = kv.getValue().toString().replace("partdate" + "=", "");
//                if (!retainPartition.contains(partitionValue)) {
//                    partitionStrList.add(partitionValue);
//                }
//            }
//        }

        log.info("删除分区：" + partitionStrList);

        if(partitionStrList.size() == 0){
            log.info("没有需要删除的分区。。。。");
            return;
        }
        HiveMetaStoreClient hiveMetaStoreClient = initHiveMetaStoreClient(hiveMetastoreUri);

        delHivePartitionData(hiveMetaStoreClient,dbName,tableName, partitionStrList);

        closeHiveMetaStore(hiveMetaStoreClient);

    }

}
